package data.platform.cassandra.internal.query;

import data.platform.cassandra.domain.QueryTime;
import data.platform.cassandra.infra.persistence.repository.CassandraRulePointRepository;
import data.platform.common.service.query.RulePointQueryService;
import data.platform.common.stream.aggregates.RulePoint;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.GroupedFlux;
import reactor.core.scheduler.Schedulers;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.List;

@ConditionalOnBean(name = "cassandraConfig")
@Service
@Slf4j
@RequiredArgsConstructor
public class CassandraRulePointQueryServiceImpl implements RulePointQueryService {

    @Value("${query.thread}")
    private int queryThreadNum;

    final CassandraRulePointRepository cassandraRulePointRepository;

    private static final String QUERY_SQL = "SELECT * FROM rule_point where rule_id='%s' and day='%s' and offset >='%s' and offset <='%s'";

    private static final DateTimeFormatter timeFormatter = DateTimeFormatter.ofPattern("HH:mm:ss.SSS");

    private static final DateTimeFormatter dayFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd");

    @Override
    public Flux<RulePoint> query(LocalDateTime beginTime, LocalDateTime endTime, String ruleId) {
        // 过滤查询开始时间，查询结束时间
        List<QueryTime> queryTimes = QueryTime.getQueryTimeSpan(beginTime, endTime);
        Flux<String> sqlFlux = Flux.fromIterable(queryTimes)
                .map(queryTime -> String.format(QUERY_SQL, ruleId, queryTime.getDay(), timeFormatter.format(queryTime.getStartOffSet()), timeFormatter.format(queryTime.getEndOffSet())));

        Flux<GroupedFlux<Integer, String>> groups = sqlFlux
                .groupBy(sql -> sql.hashCode() % queryThreadNum)
                .publishOn(Schedulers.boundedElastic());

        return cassandraRulePointRepository.find(groups)
                .map(eo -> {
                    RulePoint rulePoint = new RulePoint();
                    rulePoint.setRuleId(eo.getRulePointKey().getRuleId());
                    rulePoint.setEventTime(LocalDateTime.of(eo.getRulePointKey().getPartition(),eo.getRulePointKey().getOffset()));
                    rulePoint.setHitNumber(eo.getHitNumber());
                    rulePoint.setHitData(eo.getHitData());
                    return rulePoint;
                });
    }
}
